package cn.zd.demo.flink.statebackend;

import cn.zd.demo.flink.statebackend.dto.CustInfoDTO;
import com.alibaba.fastjson2.JSON;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class MyRichWindowFunction extends RichWindowFunction<CustInfoDTO, CustInfoDTO, String, TimeWindow> {
    private MapState<String, CustInfoDTO> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, CustInfoDTO>("MyMapState", String.class, CustInfoDTO.class));
    }

    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CustInfoDTO> iterable, Collector<CustInfoDTO> collector) throws Exception {
        iterable.forEach(custInfoDTO -> {
            try {
                System.out.println("["+mapState+"]接收到的数据：" + JSON.toJSONString(custInfoDTO));
                CustInfoDTO dto = null;

                dto = mapState.get(custInfoDTO.getCustNo());

                // 数据匹配 已有数据
                if (dto != null) {
                    System.out.println("数据匹配到已有数据：" + JSON.toJSONString(dto) + "\t" + JSON.toJSONString(custInfoDTO));
                    // 数据融合
                    dto.combine(custInfoDTO);
                    if(dto.isComplate()){
                        System.out.println("数据合并完成：" + JSON.toJSONString(dto));
                        collector.collect(dto);
                        mapState.remove(dto.getCustNo());
                    }else {
                        System.out.println("数据合未完成：" + JSON.toJSONString(dto));
                        mapState.put(custInfoDTO.getCustNo(), custInfoDTO);
                    }
                } else {
                    System.out.println("新数据："+JSON.toJSONString(custInfoDTO));
                    mapState.put(custInfoDTO.getCustNo(), custInfoDTO);
                }
            } catch (Exception e) {
                System.out.println("数据处理错误！" + s);
                e.printStackTrace();
            }
        });
    }
}
